package com.springboot.kafka.admin;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Author: TongRui乀
 * @Date: 2020-07-19 9:55
 * @description：  Kafka Admin API
 */
@Slf4j
public class TestAdmin {

    private static AdminClient adminClient;

    private static final String TOPIC_NAME = "springboot-kafka";

    static {
        Properties properties = new Properties();

        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.199.128:9092");

        adminClient = AdminClient.create(properties);

        log.info("Kafka Client initialized");
    }

    public static void main(String[] args) {
        // 创建topic
        createTopic(TOPIC_NAME);

        // 查询topic
        listTopics();

    }

    /**
     *  创建Topic
     * @param topicName
     */
    private static void createTopic(String topicName){

        adminClient.createTopics(Arrays.asList(new NewTopic(topicName, 1, (short) 1)));

        log.info("the topic {} is created", topicName);
    }

    /**
     *  查询所有的topic
     */
    private static void listTopics(){

        ListTopicsResult listTopics = adminClient.listTopics();

        try {
            listTopics.names().get().stream().forEach(System.out::println);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }


    }

}
